package com.zxl.study.service.impl;

import com.alibaba.fastjson.JSON;
import com.zxl.study.config.CacheManager;
import com.zxl.study.entities.MetaDatabase;
import com.zxl.study.entities.MetaTable;
import com.zxl.study.model.vo.MetaColumnByTableIdVo;
import com.zxl.study.model.vo.MetaColumnByTableListVo;
import com.zxl.study.model.vo.MetaDatabaseTreeVo;
import com.zxl.study.service.DatabaseService;
import com.zxl.study.service.MetaDatabaseService;
import com.zxl.study.support.BeanHelper;
import com.zxl.study.support.ClientContext;
import com.zxl.study.support.CompressionUtils;
import com.zxl.study.support.Constants;
import io.micrometer.core.instrument.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class DatabaseServiceImpl implements DatabaseService {
    private static final int CHUNK_SIZE = 100; // 每个小键的大小
    @Resource
    private CacheManager cacheManager;
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private ClientContext<?> clientContext;
    @Resource
    @Lazy
    private MetaDatabaseService metaDatabaseService;


    /**
     * 获取数据库下的所有表信息
     *
     * @param metaDatabase 待获取数据库信息的数据库连接信息
     * @return List<MetaTable> 数据库下的所有表信息
     */
    @Override
    public List<MetaTable> getRemoteTables(MetaDatabase metaDatabase) {
        return clientContext.getRemoteTables(metaDatabase);
    }


    /**
     * 获取数据库下的所有表信息,按照模式名进行划分
     *
     * @param metaDatabase 待获取数据库信息的数据库连接信息
     * @return MetaDatabaseTreeVo 数据库下的所有表信息
     */
    @Override
    public MetaDatabaseTreeVo getMetaDatabaseTablesData(MetaDatabase metaDatabase) throws IOException {
        MetaDatabaseTreeVo metaDatabaseTreeVo = readRedisData(metaDatabase);
        if (metaDatabaseTreeVo.getName() == null && metaDatabaseTreeVo.getMetaColumnByTableListVos() == null) {
            List<MetaTable> remoteTables = getRemoteTables(metaDatabase);
            List<MetaColumnByTableListVo> resultList = buildMetaDatabaseTree(remoteTables);
            metaDatabaseTreeVo.setName(metaDatabase.getName());
            metaDatabaseTreeVo.setMetaColumnByTableListVos(resultList);
            saveRedisData(metaDatabaseTreeVo, metaDatabase);
        }
        return metaDatabaseTreeVo;
    }

    /**
     * 构建当前数据库下的表的节点信息
     *
     * @param remoteTables 数据库的表信息
     * @return List<MetaColumnByTableListVo> 当前数据库下的表的节点信息
     */
    private List<MetaColumnByTableListVo> buildMetaDatabaseTree(List<MetaTable> remoteTables) {
        List<MetaColumnByTableIdVo> nodeInfos = convertToColumnVos(remoteTables);
        List<String> schemaNames = extractSchemaNames(remoteTables);

        List<MetaColumnByTableListVo> resultList = new ArrayList<>();
        List<MetaColumnByTableIdVo> metaColumnByTableIdVoListForNull = new ArrayList<>();

        for (String schemaName : schemaNames) {
            MetaColumnByTableListVo metaColumnByTableListVo = new MetaColumnByTableListVo();
            List<MetaColumnByTableIdVo> metaColumnByTableIdVoList = new ArrayList<>();
            for (MetaColumnByTableIdVo nodeInfo : nodeInfos) {
                if (schemaName != null && schemaName.equals(nodeInfo.getSchemaName())) {
                    metaColumnByTableIdVoList.add(nodeInfo);
                } else if (nodeInfo.getSchemaName() == null) {
                    metaColumnByTableIdVoListForNull.add(nodeInfo);
                }
            }
            if (schemaName != null) {
                metaColumnByTableListVo.setColumnList(metaColumnByTableIdVoList);
                metaColumnByTableListVo.setSchemaName(schemaName);
                resultList.add(metaColumnByTableListVo);
            }
        }

        if (!metaColumnByTableIdVoListForNull.isEmpty()) {
            MetaColumnByTableListVo metaColumnByTableListVo = new MetaColumnByTableListVo();
            metaColumnByTableListVo.setColumnList(metaColumnByTableIdVoListForNull);
            resultList.add(metaColumnByTableListVo);
        }
        return resultList;
    }

    /**
     * 将表实体转换成前端展示用的节点信息
     *
     * @param metaTables 待转换的metaTable列表
     * @return List<MetaColumnByTableIdVo> 前端展示用的节点信息
     */
    private List<MetaColumnByTableIdVo> convertToColumnVos(List<MetaTable> metaTables) {
        return metaTables.stream().map(metaTable -> {
            MetaColumnByTableIdVo columnVo = BeanHelper.copyProperties(metaTable, MetaColumnByTableIdVo.class);
            columnVo.setTreeType(Constants.META_TREE_TYPE_TABLE);
            columnVo.setTreeId(Constants.META_TREE_TYPE_TABLE + columnVo.getId());
            String label = StringUtils.isNotBlank(metaTable.getNameAlias()) ? metaTable.getNameAlias() : metaTable.getName();
            columnVo.setLabel(label);
            columnVo.setLeaf(true);
            columnVo.setDisabled(false);
            return columnVo;
        }).collect(Collectors.toList());
    }

    /**
     * 从表信息中过滤出模式信息
     *
     * @param metaTables 待过滤的metaTable列表
     * @return List<MetaColumnByTableIdVo> 模式信息
     */
    private List<String> extractSchemaNames(List<MetaTable> metaTables) {
        return metaTables.stream().map(MetaTable::getSchemaName).distinct().collect(Collectors.toList());
    }

    /**
     * 将metaDatabaseTreeVo拆分存储入缓存
     *
     * @param metaDatabaseTreeVo 待保存的 metaDatabaseTreeVo
     * @param metaDatabase       待保存的库信息
     * @throws IOException 压缩数据的异常
     */
    public void saveRedisData(MetaDatabaseTreeVo metaDatabaseTreeVo, MetaDatabase metaDatabase) throws IOException {
        String keyPrefixWithDb = "database::" + metaDatabase.getId();
        metaDatabaseService.scanRedisData(metaDatabase);
        for (MetaColumnByTableListVo tableListVo : metaDatabaseTreeVo.getMetaColumnByTableListVos()) {
            String schemaName = tableListVo.getSchemaName();
            String columnKeyPrefix = keyPrefixWithDb + "::" + schemaName + "::chunk::";
            List<MetaColumnByTableIdVo> columnList = tableListVo.getColumnList();
            saveInChunks(columnKeyPrefix, columnList);
        }
    }

    /**
     * 将metaDatabaseTreeVo的表节点list拆分成小key，分别存入缓存
     *
     * @param keyPrefix 小key的前缀
     * @param dataList  待拆分数据
     * @throws IOException 压缩数据的异常
     */
    private void saveInChunks(String keyPrefix, List<MetaColumnByTableIdVo> dataList) throws IOException {
        int totalChunks = (int) Math.ceil((double) dataList.size() / CHUNK_SIZE);
        for (int i = 0; i < totalChunks; i++) {
            int start = i * CHUNK_SIZE;
            int end = Math.min(start + CHUNK_SIZE, dataList.size());
            List<MetaColumnByTableIdVo> sublist = dataList.subList(start, end);
            String sublistKey = keyPrefix + i;
            String compressedData = CompressionUtils.compress(JSON.toJSONString(sublist));
            redisTemplate.opsForValue().set(sublistKey, compressedData, 1, TimeUnit.HOURS);
        }
    }

    /**
     * 获取指定数据库的metaDatabaseTreeVo
     *
     * @param metaDatabase 待读取redis数据的库信息
     * @return MetaDatabaseTreeVo  指定数据库的metaDatabaseTreeVo
     */
    public MetaDatabaseTreeVo readRedisData(MetaDatabase metaDatabase) {
        MetaDatabaseTreeVo metaDatabaseTreeVo = new MetaDatabaseTreeVo();
        List<MetaColumnByTableListVo> metaColumnByTableListVos = new ArrayList<>();
        String databaseKeyPrefix = "database::" + metaDatabase.getId();
        ScanOptions options = ScanOptions.scanOptions().match(databaseKeyPrefix + "*::chunk::*").count(100).build();
        Cursor<byte[]> cursor = (Cursor<byte[]>) redisTemplate.executeWithStickyConnection(redisConnection -> redisConnection.scan(options));

        assert cursor != null;
        cursor.forEachRemaining(rawKey -> {
            try {
                processKey(new String(rawKey), databaseKeyPrefix, metaColumnByTableListVos);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        if (!metaColumnByTableListVos.isEmpty()) {
            metaDatabaseTreeVo.setName(metaDatabase.getName());
            metaDatabaseTreeVo.setMetaColumnByTableListVos(metaColumnByTableListVos);
        }
        return metaDatabaseTreeVo;
    }

    private void processKey(String key, String databaseKeyPrefix, List<MetaColumnByTableListVo> metaColumnByTableListVos) throws IOException {
        String schemaName =key.substring(databaseKeyPrefix.length() + 2, key.indexOf("::chunk::"));
        MetaColumnByTableListVo metaColumnByTableListVo = findOrCreateSchemaList(schemaName, metaColumnByTableListVos);
        List<MetaColumnByTableIdVo> sublist = fetchAndDecompressData(key);

        if (sublist != null) {
            metaColumnByTableListVo.getColumnList().addAll(sublist);
        }
    }

    /**
     * 读取指定key的缓存数据
     *
     * @param key 缓存key
     * @return List<MetaColumnByTableIdVo>  指定key的缓存数据
     */
    private List<MetaColumnByTableIdVo> fetchAndDecompressData(String key) throws IOException {
        String compressedData = (String) cacheManager.getCacheFrontend().get(key);
        if (compressedData != null) {
            String jsonData = compressedData.substring(1, compressedData.length() - 1).replace("\\\"", "\"");
            jsonData = CompressionUtils.decompress(jsonData);
            return JSON.parseArray(jsonData, MetaColumnByTableIdVo.class);
        }
        return null;
    }

    /**
     *
     * @param schemaName 模式名称
     * @param columnByTableListVos 表信息
     * @return MetaColumnByTableListVo
     */
    private MetaColumnByTableListVo findOrCreateSchemaList(String schemaName, List<MetaColumnByTableListVo> columnByTableListVos) {
        for (MetaColumnByTableListVo tableListVo : columnByTableListVos) {
            if (tableListVo.getSchemaName().equals(schemaName)) {
                return tableListVo;
            }
        }
        MetaColumnByTableListVo newTableListVo = new MetaColumnByTableListVo();
        newTableListVo.setSchemaName(schemaName);
        newTableListVo.setColumnList(new ArrayList<>());
        columnByTableListVos.add(newTableListVo);
        return newTableListVo;
    }
}
